package project.p1_car

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig

import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.sql.{Connection, DriverManager, ResultSet, Statement}


/**
 * 需求：根据卡口上报的实时信息（消费kafka）结合mysql中的限速信息监控超速车辆并将结果存储mysql
 *
 */
object D1_monitor_out_of_speed_car {


  val TOPIC:String = "flink_traffic_log"

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092")
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink_group_1")

    val kafkaConsumer = new FlinkKafkaConsumer[String](TOPIC, new SimpleStringSchema(), prop)
    kafkaConsumer.setStartFromLatest()


    val trafficLogStream: DataStream[TrafficInfo] = env.addSource(kafkaConsumer)
      .map(line => {
        val arrs: Array[String] = line.split(",")
        new TrafficInfo(arrs(0).toLong, arrs(1), arrs(2), arrs(3), arrs(4).toDouble, arrs(5), arrs(6))
      })


    val speedMonitorDesc = new MapStateDescriptor[String, MonitorInfo]("speed_monitor", classOf[String], classOf[MonitorInfo])


    val monitorSpeedBcStream: BroadcastStream[MonitorInfo] = env.addSource(new JdbcSource())
      .broadcast(speedMonitorDesc)


    val outOfSpeedStream: DataStream[OutOfSpeedingInfo] = trafficLogStream.connect(monitorSpeedBcStream)
      .process(new BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfSpeedingInfo] {

        override def processElement(trafficInfo: TrafficInfo, ctx: BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfSpeedingInfo]#ReadOnlyContext, out: Collector[OutOfSpeedingInfo]): Unit = {

          val info: MonitorInfo = ctx.getBroadcastState(speedMonitorDesc).get(trafficInfo.monitor_id)

          if (info != null && trafficInfo.speed > info.speedLimit * 1.1) {
            val speedInfo = new OutOfSpeedingInfo(trafficInfo.car,
              trafficInfo.monitor_id,
              trafficInfo.roadId,
              trafficInfo.speed,
              info.speedLimit,
              trafficInfo.action_time)


            out.collect(speedInfo)

          }

        }

        override def processBroadcastElement(monitorInfo: MonitorInfo, ctx: BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfSpeedingInfo]#Context, out: Collector[OutOfSpeedingInfo]): Unit = {
          ctx.getBroadcastState(speedMonitorDesc).put(monitorInfo.monitor_id, monitorInfo)
        }
      })


    outOfSpeedStream.print()



    env.execute("watchOutOfSpeedApp")




  }

}




class JdbcSource extends RichSourceFunction[MonitorInfo] {


  var flag = true

  var conn:Connection = _

  var statement:Statement = _

  var resultSet:ResultSet = _


  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://node1:3306/db_flink", "root", "123456")

    statement = conn.createStatement()

  }

  override def run(ctx: SourceFunction.SourceContext[MonitorInfo]): Unit = {

    while (flag) {
      resultSet = statement.executeQuery("SELECT * FROM db_flink.t_monitor_info")

      while(resultSet.next()) {

        val info = new MonitorInfo(resultSet.getString(1),
          resultSet.getString(2),
          resultSet.getInt(3),
          resultSet.getString(4))

        ctx.collect(info)
      }
    }

  }

  override def cancel(): Unit = {
    flag = false
  }


  override def close(): Unit = {
    statement.close()
    conn.close()
  }
}



class JdbcSink extends RichSinkFunction[OutOfSpeedingInfo] {
  override def open(parameters: Configuration): Unit = {


  }
}